package spark.core.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object rddCheckPoint {
  // Todo RDD CheckPoint 检查点
  // 缓存和检查点区别：
  /**
   * 1）Cache 缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
   * 2）Cache 缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。
   * 3）建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。
   * */
  def main(args: Array[String]): Unit = {
    // 配置spark上下文
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rddPersistence")
    val sc: SparkContext = new SparkContext(sparkConf)
    // 设置检查点路径
    sc.setCheckpointDir("checkPoint")

    // 创建一个RDD，读取指定位置文件
    val lineRdd: RDD[String] = sc.textFile("Input/test.txt")

    val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    val wordToOneRdd: RDD[(String, Long)] = wordRdd.map {
      word => {
        (word, System.currentTimeMillis())
      }
    }
    // 增加缓存,避免再重新跑一个job做checkpoint
    // 原因：这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。
    wordToOneRdd.cache()
    // 数据检查点：针对wordToOneRdd做检查点计算
    wordToOneRdd.checkpoint()
    // 触发执行逻辑
    wordToOneRdd.collect().foreach(println)

  }

}
